package NIO;

import com.sun.security.ntlm.Server;
import sun.tools.jconsole.Worker;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.SocketAddress;
import java.net.SocketOption;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import static NIO.ByteBufferUtil.debugAll;

//利用多线程的机制改进服务端，充分利用cpu
//让主线程管理连接，并且创建新的线程管理对应socketChannel上的数据读写

public class Server_multiThreading {
    public static void main(String[] args) throws IOException {
            new BossEventLoop().register();
    }
    static class BossEventLoop implements Runnable{
        private Selector boss;//用来管理accept事件的
        private WorkerEventLoop[] workers;
        private volatile boolean start=false;
        AtomicInteger index=new AtomicInteger();//原子类，在分配worker的时候利用它实现负载均衡

        public void register() throws IOException {
            if (!start){
                ServerSocketChannel ssc=ServerSocketChannel.open();
                ssc.bind(new InetSocketAddress(8080));
                ssc.configureBlocking(false);
                boss=Selector.open();
                SelectionKey key = ssc.register(boss, 0, null);
                key.interestOps(SelectionKey.OP_ACCEPT);
                workers=initEventLoops();
                new Thread(this,"boss").start();
                System.out.println("boss start");
                start=true;
            }
        }

        public WorkerEventLoop[] initEventLoops(){
            WorkerEventLoop[] workerEventLoops=new WorkerEventLoop[2];
            //可以根据主机的cpu数量进行设定worker的数量
            for (int i=0;i< workerEventLoops.length;i++)
                workerEventLoops[i]=new WorkerEventLoop(i);
            return workerEventLoops;
        }
        @Override
        public void run() {
            while (true){
                try {
                    //只负责accept事件，接收连接
                    boss.select();
                    Iterator<SelectionKey> iterator = boss.keys().iterator();
                    while (iterator.hasNext()){
                        SelectionKey key= iterator.next();
                        if (key.isAcceptable()){
                            ServerSocketChannel channel =(ServerSocketChannel) key.channel();
                            //获取得到的socketchannel
                            SocketChannel sc = channel.accept();
                            sc.configureBlocking(false);
                            System.out.println("connnected:"+sc.getRemoteAddress());
                            //将sockerchannel注册到worker上，让worker负责这个sc的接收
                            workers[index.getAndIncrement()% workers.length].register(sc);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        }
    }
    static class WorkerEventLoop implements Runnable{
        private Selector worker;
        private volatile  boolean start=false;
        private int index;

        //模仿netty
        private final ConcurrentLinkedQueue<Runnable> tasks=new ConcurrentLinkedQueue<>();

        public WorkerEventLoop(int index){
            this.index=index;
        }

        public void register(SocketChannel sc) throws IOException {
            if (!start){
                worker=Selector.open();
                new Thread(this,"worker"+index).start();
                start=true;
            }

            //因为在select方法阻塞的时候，调用sc.register会不好使，所以需要先将其放到队列中等待
            //其实有很多种方式可以用，这里只是模拟netty的实现
            tasks.add(()->{
                try {
                    SelectionKey key= sc.register(worker, 0, null);
                    key.interestOps(SelectionKey.OP_READ);
                    worker.selectNow();
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });

            //让select方法不阻塞，接着运行
            worker.wakeup();
        }

        @Override
        public void run() {
            while (true){
                try {
                    worker.select();
                    //select不阻塞后，sc.register方法可以运行，所以此时要判断队列中有没有需要进行register的事件，如果有的话，执行
                    Runnable task= tasks.poll();
                    if (task!=null)
                        task.run();
                    Iterator<SelectionKey> iter = worker.keys().iterator();
                    while (iter.hasNext()){
                        SelectionKey key = iter.next();
                        if (key.isReadable()){
                            SocketChannel sc =(SocketChannel) key.channel();
                            ByteBuffer buffer=ByteBuffer.allocate(16);
                            try {
                                int read = sc.read(buffer);
                                if (read == -1) {
                                    //代表没有数据可读了，所以要cancel，和断开连接
                                    key.cancel();
                                    sc.close();
                                } else {
                                    //切换读模式
                                    buffer.flip();
                                    System.out.println("Message----------" + sc.getRemoteAddress());
                                    debugAll(buffer);
                                }
                            }catch (IOException e){
                                //如果此时客户端断开连接了，会出现异常，所以此时也需要cancel和close
                                e.printStackTrace();
                                key.cancel();
                                sc.close();
                            }
                        }

                    }
                } catch (IOException e) {
                    e.printStackTrace();

                }

            }
        }
    }
}
